package com.bluemoon.daps.test.sync.v1_3.devimport;

import cn.com.bluemoon.daps.common.toolkit.DbTypeHelper;
import cn.com.bluemoon.daps.common.enums.DatabaseType;
import cn.com.bluemoon.daps.common.enums.SyncToEnv;
import cn.com.bluemoon.daps.common.exception.DapThrowException;
import cn.com.bluemoon.daps.common.sql.SqlScriptUtils;
import cn.com.bluemoon.daps.common.toolkit.BmAssetUtils;
import cn.com.bluemoon.daps.model.dto.ModelTableDto;
import cn.com.bluemoon.daps.model.entity.*;
import cn.com.bluemoon.daps.sync.core.sync.GlobalConfig;
import cn.com.bluemoon.daps.sync.core.sync.TableModel;
import cn.com.bluemoon.daps.sync.core.sync.meta.MetadataConverter;
import cn.com.bluemoon.daps.sync.feignclient.DapModelModuleFacade;
import cn.com.bluemoon.daps.sync.feignclient.DapSystemModuleFacade;
import cn.com.bluemoon.daps.system.entity.DapSystemConfigSandbox;
import cn.com.bluemoon.daps.system.entity.DapSystemDatasource;
import cn.com.bluemoon.daps.system.entity.DapSystemDatasourceEnvironment;
import cn.com.bluemoon.daps.system.entity.DapSystemInfo;
import cn.com.bluemoon.metadata.common.ResultBean;
import cn.com.bluemoon.metadata.common.enums.DbTypeEnum;
import cn.com.bluemoon.metadata.inter.DbMetaDataService;
import cn.com.bluemoon.metadata.inter.dto.in.QueryMetaDataRequest;
import cn.com.bluemoon.metadata.inter.dto.out.ColumnInfoVO;
import cn.com.bluemoon.metadata.inter.dto.out.SchemaInfoVO;
import cn.com.bluemoon.metadata.inter.dto.out.SqlIndexInfoVO;
import cn.com.bluemoon.metadata.inter.dto.out.TableInfoVO;
import com.alibaba.druid.DbType;
import lombok.extern.slf4j.Slf4j;
import org.junit.Before;
import reactor.util.function.Tuple2;
import reactor.util.function.Tuple4;
import reactor.util.function.Tuples;

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.*;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.Collectors;

import static cn.hutool.core.text.CharSequenceUtil.trim;

/**
 * @author Jarod.Kong
 * @date 2021/6/9 9:21
 */
@Slf4j
public abstract class BaseBefore {
    protected GlobalConfig globalConfig;
    protected SyncToEnv syncToEnv = SyncToEnv.MODEL;
    @Resource
    private DapModelModuleFacade modelService;
    @Resource
    private DapSystemModuleFacade systemModuleFacade;
    @Resource
    private DbMetaDataService dbMetaDataService;

    public BaseBefore(SyncToEnv syncToEnv) {
        this.syncToEnv = syncToEnv;
    }

    public void setSyncToEnv(SyncToEnv syncToEnv) {
        this.syncToEnv = syncToEnv;
    }

    @Before
    public void beforeClass() {
        String modelId;
        modelId = "1419549042953445378";
        //  2021/6/9 需要根据最新的数据源业务进行调整！！ 每个同步要根据对应的同步环境进行获取连接信息
        DapDataModel model = modelService.getByIdAsset(modelId, "模型不存在！！");
        Optional<DapSystemInfo> systemInfo = systemModuleFacade.getSystemInfo(model.getSystemId());
        BmAssetUtils.isTrue(systemInfo.isPresent(), "获取系统信息失败！！");
        //  2021/6/16 根据syncToEnv数据源环境+数据源id获取源数据信息
//        DapSystemDatasource ds = systemModuleFacade.getDbTypeByDataSourceId(model.getDataSourceId());
        GlobalConfig.DataSourceAndEnv dsDto = systemModuleFacade.getSysDataSourceAndEnvByIdAndEnv(model.getDataSourceId(), syncToEnv);
        DapSystemDatasource ds = dsDto.getDs();
        BmAssetUtils.isTrue(DatabaseType.MYSQL.equals(ds.getDatabaseType()), "该功能仅支持MySQL数据库");
        List<DapDataModelGroup> defaultGroups = modelService.initModelDefaultGroup(modelId);
        BmAssetUtils.isFalse(defaultGroups.isEmpty(), "模型不存在默认组！！");
        DapDataModelGroup defaultGroup = defaultGroups.get(0);
        // global
        GlobalConfig globalConfig = GlobalConfig.builder()
                .model(model)
                .dataSourceAndEnv(dsDto)
                .addSink(false)
                .dbType(DbTypeHelper.getDruidDbType(ds.getDatabaseType(), DbType.mysql))
                .system(systemInfo.get())
                .syncToEnv(syncToEnv)
                .defaultGroup(defaultGroup).build();
        // TODO: 2021/6/16 根据最新数据源设计调整该赋值
        globalConfig.getDataSource().setIsPtosc(true); // 单元测试需要
        globalConfig.getDataSource().setIsSync(true); // 单元测试需要
        Boolean isSync = globalConfig.getDataSource().getIsSync();
        BmAssetUtils.isTrue(isSync, "模型未开启同步功能！！");
        this.globalConfig = globalConfig;
    }


    /**
     * 获取模型信息
     *
     * @return
     */
    public List<ModelTableDto> getLocalModelTableDtos() {
        Tuple4<List<DapDataModelTable>, List<DapDataModelTableField>,
                List<DapDataModelSubordinateKey>, List<DapDataModelTablePartition>>
                modelInfos = modelService.queryModelInfosByModelAndNewestAndTableIds(globalConfig.getModel().getId(), globalConfig.isModelNewest(), globalConfig.getTableIdentity().getTableIdsDistinct());
        log.info("获取到模型表：{}，字段：{}，索引：{}，分区：{}", modelInfos.getT1().size(), modelInfos.getT2().size(),
                modelInfos.getT3().size(), modelInfos.getT4().size());
        log.info("开始整合表、字段、索引信息");
        Map<String, List<DapDataModelTableField>> tableNameAndFields = modelInfos.getT2().stream().collect(Collectors.groupingBy(DapDataModelTableField::getTableName));
        Map<String, DapDataModelTableField> fieldIdAndField = modelInfos.getT2().stream().collect(Collectors.toMap(DapDataModelTableField::getId, a -> a, (a, b) -> a));
        Map<String, List<DapDataModelSubordinateKey>> tableNameAndIndexes = modelInfos.getT3().stream().collect(Collectors.groupingBy(DapDataModelSubordinateKey::getTableName));
        List<ModelTableDto> collect = modelInfos.getT1().stream().map(t -> {
            List<DapDataModelSubordinateKey> indexes = tableNameAndIndexes.getOrDefault(t.getEngName(), Collections.emptyList());
            indexes.forEach(i -> {
                List<DapDataModelTableField> indexFields = Arrays.stream(i.getFieldIdsSplit())
                        .filter(fieldIdAndField::containsKey).map(fieldIdAndField::get).collect(Collectors.toList());
                i.setIndexFields(indexFields);
                i.setFieldNames(indexFields.stream().map(DapDataModelTableField::getFieldName).collect(Collectors.joining(DapDataModelSubordinateKey.INDEX_FIELDIDS_CHAR)));
            });
            return ModelTableDto.builder().schemaName(t.getTableSchema())
                    .table(t)
                    .fields(tableNameAndFields.getOrDefault(t.getEngName(), Collections.emptyList()))
                    .indexes(indexes)
                    .partitions(Collections.emptyList())
                    .build();
        }).collect(Collectors.toList());
        return collect;
    }

    public SchemaInfoVO getSchemaInfoVO(GlobalConfig.DataSourceAndEnv dataSourceAndEnv) {
        DapSystemDatasource ds = dataSourceAndEnv.getDs();
        DapSystemDatasourceEnvironment env = dataSourceAndEnv.getEnv();
        QueryMetaDataRequest req = new QueryMetaDataRequest();
        req.setDbName(env.getDatabaseName());
        req.setIp(env.getHost());
        req.setPort(String.valueOf(env.getPort()));
        req.setPassword(env.decryptPassword());
        req.setUsername(env.getUsername());
        DatabaseType dbType = ds.getDatabaseType();
        BmAssetUtils.isTrue(DatabaseType.MYSQL.equals(dbType), "支持MySQL数据库");
        req.setDbType(DbTypeEnum.MYSQL);
        req.setSchemaName(env.getDatabaseName());
        ResultBean<SchemaInfoVO> schemaInfoVOResultBean = dbMetaDataService.queryMetaData(req);
        BmAssetUtils.isTrue(schemaInfoVOResultBean.getCode() == 200, schemaInfoVOResultBean
                .getMsg());
        return schemaInfoVOResultBean.getContent();
    }

    /**
     * 获取沙箱环境的元数据
     *
     * @param globalConfig g
     * @return 获取沙箱环境的元数据
     */
    public SchemaInfoVO getSandboxSchemaInfoVO(GlobalConfig globalConfig) {
        GlobalConfig.DataSourceAndEnv dataSourceAndEnv = globalConfig.getDataSourceAndEnv();
        DapSystemDatasource ds = globalConfig.getDataSource();
        DapSystemDatasourceEnvironment env = dataSourceAndEnv.getEnv();
        QueryMetaDataRequest req = new QueryMetaDataRequest();
        DapSystemConfigSandbox sandbox = globalConfig.getDsSandboxEnv();
        req.setPassword(sandbox.decryptPassword());
        req.setUsername(sandbox.getUsername());
        req.setIp(sandbox.getHost());
        req.setPort(String.valueOf(sandbox.getPort()));
        log.warn("当前配置为同步到沙箱环境中");
        req.setDbName(env.getDatabaseName());
        DatabaseType dbType = ds.getDatabaseType();
        BmAssetUtils.isTrue(DatabaseType.MYSQL.equals(dbType), "支持MySQL数据库");
        req.setDbType(DbTypeEnum.MYSQL);
        req.setSchemaName(env.getDatabaseName());
        ResultBean<SchemaInfoVO> schemaInfoVOResultBean = dbMetaDataService.queryMetaData(req);
        BmAssetUtils.isTrue(schemaInfoVOResultBean.getCode() == 200, schemaInfoVOResultBean
                .getMsg());
        return schemaInfoVOResultBean.getContent();
    }


    private Tuple2<String, List<ModelTableDto>> getMetadataDataModelTableDtos() {
        SchemaInfoVO schemaInfoVO = getSchemaInfoVO(globalConfig.getDataSourceAndEnv());
        String databaseName = globalConfig.getDataSourceEnv().getDatabaseName();
        if (schemaInfoVO.getName() == null) {
            log.warn("{}环境下，当前数据库不存在该数据库名{}或者表信息", globalConfig.getSyncToEnv().getDesc(), databaseName);
            boolean status = globalConfig.initDbName(false);
            log.warn(String.format("初始化数据库名%s情况：%s", databaseName, status));
            return null;
        }
        if (schemaInfoVO.getTables() == null || schemaInfoVO.getTables().isEmpty()) {
            log.warn("{}环境下，当前数据库不存在该数据库名{}或者表信息", globalConfig.getSyncToEnv().getDesc(), databaseName);
            return null;
        }
//        BmAssetUtils.notNull(schemaInfoVO.getName(), "元数据获取数据库schema/dbname为空！");
        List<TableInfoVO> tables = schemaInfoVO.getTables();
        List<ModelTableDto> tableDtos = tables.stream().map(t -> {
            List<ColumnInfoVO> columns = t.getColumns();
            List<SqlIndexInfoVO> sqlIndexes = t.getSqlIndexes();
            return ModelTableDto.builder()
                    .schemaName(schemaInfoVO.getName())
                    .table(MetadataConverter.Table.mgToLocal(t))
                    .fields(columns.stream().map(MetadataConverter.Field::mgToLocal).filter(Objects::nonNull).collect(Collectors.toList()))
                    .indexes(sqlIndexes.stream().map(MetadataConverter.Index::mgToLocal).filter(Objects::nonNull).collect(Collectors.toList()))
                    .partitions(Collections.emptyList())
                    .build();
        }).collect(Collectors.toList());
        return Tuples.of(schemaInfoVO.getName(), tableDtos);
    }

    /**
     * @return 获取模型对应的开发库的数据集
     */
    public List<TableModel> getMetadataTableModel(Set<String> tableIds) {
        List<TableModel> tableModels = getMetadataTableModel();
        List<TableModel> metadataTables = tableModels.stream().filter(d -> new ToTableModelPredicate(tableIds).test(d)).collect(Collectors.toList());
        log.info("获取选中的查询元数据的表，{}, size={}", metadataTables, metadataTables.size());
        return metadataTables;
    }

    /**
     * @return 获取模型对应的开发库的数据集
     */
    public List<TableModel> getMetadataTableModel() {
        Tuple2<String, List<ModelTableDto>> dbAndTables = getMetadataDataModelTableDtos();
        if (dbAndTables == null) {
            return Collections.emptyList();
        }
        List<ModelTableDto> metadataDataModelTableDtos = null;
        String dbNameSchema;
        metadataDataModelTableDtos = dbAndTables.getT2();
        dbNameSchema = dbAndTables.getT1();
        final String finalDbNameSchema = trim(dbNameSchema);
        List<TableModel> modelV2 = metadataDataModelTableDtos.stream().map(new ToTableModelFun(finalDbNameSchema)).collect(Collectors.toList());
        return modelV2;
    }

    /**
     * @return 获取本地模型的数据集
     */
    public List<TableModel> getLocalTableModel() {
        List<ModelTableDto> localModelTableDtos = getLocalModelTableDtos();
        List<TableModel> modelV1 = localModelTableDtos.stream().map(new ToTableModelFun(null)).collect(Collectors.toList());
        return modelV1;
    }

    /**
     * @param tableIds set
     * @return 获取本地+元数据集+过滤非所选表
     */
    public Tuple2<List<TableModel>, List<TableModel>> getLocalAndMetadataTableModel(Set<String> tableIds) {
        Tuple2<List<TableModel>, List<TableModel>> tuple2 = getLocalAndMetadataTableModel();
        List<TableModel> t2 = tuple2.getT2();
        List<TableModel> tbs = t2.stream().filter(d -> new ToTableModelPredicate(tableIds).test(d)).collect(Collectors.toList());
        return Tuples.of(tuple2.getT1(), tbs);
    }

    /**
     * @param tbIds array
     * @return 获取本地+元数据集+过滤非所选表
     */
    public Tuple2<List<TableModel>, List<TableModel>> getLocalAndMetadataTableModel(String[] tbIds) {
        HashSet<String> tableIds = new HashSet<>(Arrays.asList(tbIds));
        return getLocalAndMetadataTableModel(tableIds);
    }

    /**
     * @return 获取本地+元数据集
     */
    public Tuple2<List<TableModel>, List<TableModel>> getLocalAndMetadataTableModel() {
        ExecutorService queryMts = Executors.newFixedThreadPool(2);
        Future<List<TableModel>> localFuture = queryMts.submit(this::getLocalTableModel);
        Future<List<TableModel>> metaFuture = queryMts.submit((Callable<List<TableModel>>) this::getMetadataTableModel);
        List<TableModel> modelV1, modelV2;
        try {
            modelV2 = metaFuture.get();
            modelV1 = localFuture.get();
        } catch (InterruptedException | ExecutionException e) {
            throw new DapThrowException("异步获取元数据/模型数据失败！！", e);
        } finally {
            queryMts.shutdown();
            log.info("查询元数据/模型数据结束，关闭线程池");
        }
        if (modelV1 == null) modelV1 = Collections.emptyList();
        if (modelV2 == null) modelV2 = Collections.emptyList();
        return Tuples.of(modelV1, modelV2);
    }

    /**
     * 统一处理本地+元数据结果集
     */
    public static class ToTableModelFun implements Function<ModelTableDto, TableModel> {
        private final String dbNameSchema;

        public ToTableModelFun(String dbNameSchema) {
            this.dbNameSchema = dbNameSchema;
        }

        @Override
        public TableModel apply(ModelTableDto i) {
            TableModel t = new TableModel();
            i.getTable().setTableSchema(dbNameSchema);
            i.getTable().setDbName(dbNameSchema);
            i.getTable().setBmKey(trim(i.getTable().getEngName()));
            i.getTable().setName(SqlScriptUtils.formatComment(i.getTable().getName()));
            t.setTable(i.getTable());
            i.getFields().forEach(f -> {
                // trim
                f.setBmKey(trim(f.getFieldName()));
                if ("".equals(trim(f.getCompoundType()))) {
                    f.setCompoundType(null);
                }
                f.setComment(SqlScriptUtils.formatComment(f.getComment()));
                f.setLength(trim(f.getLength()));
                f.setDataType(trim(f.getDataType()));
                f.setFieldName(trim(f.getFieldName()));
                f.setDefaultValue(trim(f.getDefaultValue()));
                // 和模型保持统一，元数据保持统一
                if (f.getDecimalLength() == null || f.getDecimalLength() == 0) {
                    f.setDecimalLength(null);
                }
                // 要是该字段为主键，则设置notNull为true
                if (f.getPrimaryKey() != null && f.getPrimaryKey() == 1) {
                    f.setNotNull(1);
                }
            });
            t.setFields(i.getFields());
            i.getIndexes().forEach(f -> {
                f.setBmKey(trim(f.getKeyName()));
                f.setFieldIds(trim(f.getFieldIds()));
                f.setFieldNames(trim(f.getFieldNames()));
                f.setKeyChinName(trim(f.getKeyChinName()));
                if (f.getKeyChinName() != null) {
                    f.setKeyChinName(SqlScriptUtils.formatComment(f.getKeyChinName()));
                }
                f.setKeyName(trim(f.getKeyName()));
            });
            t.setIndexes(i.getIndexes());
            i.getPartitions().forEach(f -> {
                f.setBmKey(trim(f.getPartitionFieldName()));
                f.setPartitionFieldName(trim(f.getPartitionFieldName()));
                f.setPartitionFieldType(f.getPartitionFieldType());
            });
            t.setPartitions(i.getPartitions());
            return t;
        }
    }

    public static class ToTableModelPredicate implements Predicate<TableModel> {
        private final Set<String> tableEngNames;

        public ToTableModelPredicate(Set<String> tableEngNames) {
            this.tableEngNames = tableEngNames;
        }


        @Override
        public boolean test(TableModel d) {
            if (tableEngNames == null) return true;
            return tableEngNames.contains(d.getTable().getEngName());
        }
    }

}
